package com.study.bigdata.spark.core.req

import org.apache.spark.{SparkConf, SparkContext}

object Spark01_Req_HotCategoryTop10 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc = new SparkContext(conf)
    // TODO 需求一：Top10热门品类\
    // TODO 问题：1.同一个RDD的重复使用 2.cogroup算子可能出现笛卡尔乘积，性能低下

    // TODO 读取文件，获取原始数据
    val fileDatas = sc.textFile("data/user_visit_action.txt")

    // TODO 统计品类的点击数量  先过滤再统计
    val clickDatas = fileDatas.filter(
      data => {
        val datas = data.split("_")
        val cid = datas(6)
        cid != "-1"
      }
    )

    val clickCntDatas = clickDatas.map(
      data => {
        val datas = data.split("_")
        val cid = datas(6)
        (cid, 1)
      }
    ).reduceByKey(_ + _)

    // TODO 统计品类的下单数量
    val orderDatas = fileDatas.filter(
      data => {
        val datas = data.split("_")
        val cid = datas(8)
        cid != "null"
      }
    )

    val orderCntDatas = orderDatas.flatMap(
      data => {
        val datas = data.split("_")
        val cid = datas(8)
        val cids = cid.split(",")
        cids.map((_, 1))
      }
    ).reduceByKey(_ + _)


    // TODO 统计品类的支付数量
    val payDatas = fileDatas.filter(
      data => {
        val datas = data.split("_")
        val cid = datas(10)
        cid != "null"
      }
    )

    val payCntDatas = payDatas.flatMap(
      data => {
        val datas = data.split("_")
        val cid = datas(10)
        val cids = cid.split(",")
        cids.map((_, 1))
      }
    ).reduceByKey(_ + _)

    // TODO 对统计结果进行排序 =>Touple(点击 下单 支付)
    val cidCntsDatas = clickCntDatas.cogroup(orderCntDatas,payCntDatas)
    val mapDatas = cidCntsDatas.map {
      case (cid, (clickIter, orderIter, payIter)) => {
        var clickcnt = 0
        var ordercnt = 0
        var paycnt = 0

        val iter1 = clickIter.iterator
        if (iter1.hasNext) {
          clickcnt = iter1.next()
        }

        val iter2 = orderIter.iterator
        if (iter2.hasNext) {
          ordercnt = iter2.next()
        }

        val iter3 = payIter.iterator
        if (iter3.hasNext) {
          paycnt = iter3.next()
        }
        (cid,(clickcnt,ordercnt,paycnt))
      }
    }
    val top10 = mapDatas.sortBy(_._2,false).take(10)


    // TODO 将结果采集打印到控制台上
//    clickCntDatas.collect().foreach(println)
//    orderCntDatas.collect().foreach(println)
//    payCntDatas.collect().foreach(println)
    top10.foreach(println)
    sc.stop()

  }

}
